package com.iogogogo.supervisord4j.service.impl;

import com.google.common.collect.Lists;
import com.iogogogo.supervisord.core.Supervisord;
import com.iogogogo.supervisord.core.rpc.SupervisordRpc;
import com.iogogogo.supervisord.domain.SupervisordProcess;
import com.iogogogo.supervisord.exception.SupervisordException;
import com.iogogogo.supervisord.properties.SupervisordProperties;
import com.iogogogo.supervisord.util.ListUtils;
import com.iogogogo.supervisord4j.base.ResultCode;
import com.iogogogo.supervisord4j.base.context.SpringContextHolder;
import com.iogogogo.supervisord4j.base.exception.BusinessException;
import com.iogogogo.supervisord4j.pojo.entity.JobEntity;
import com.iogogogo.supervisord4j.pojo.request.JobInfoRequest;
import com.iogogogo.supervisord4j.pojo.request.ReadLogRequest;
import com.iogogogo.supervisord4j.repository.SupervisordRepository;
import com.iogogogo.supervisord4j.service.SupervisordService;
import com.iogogogo.supervisord4j.util.IdHelper;
import com.iogogogo.supervisord4j.util.Java8DateTimeUtils;
import io.vavr.Tuple;
import org.apache.commons.lang3.StringUtils;
import org.jooq.lambda.Seq;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import java.io.IOException;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * Created by tao.zeng on 2021/7/7.
 */
@Service
public class SupervisordServiceImpl implements SupervisordService {

    private final static Logger LOGGER = LoggerFactory.getLogger(SupervisordServiceImpl.class);

    private final SupervisordRepository supervisordRepository;

    public SupervisordServiceImpl(SupervisordRepository supervisordRepository) {
        this.supervisordRepository = supervisordRepository;
    }

    @Override
    public void refreshUpdateTime(String instance, String processName) throws SupervisordException, IOException {
        Supervisord supervisord = supervisord(instance);
        SupervisordProcess processInfo = supervisord.getProcessInfo(processName);
        SupervisordProperties properties = supervisord.getSupervisordProperties();

        Optional<JobEntity> optional = supervisordRepository.findByInstanceAndServerAndName(instance, properties.getUrl(), processInfo.getName());

        JobEntity jobEntity = optional.orElseGet(() -> {
            JobEntity entity = new JobEntity();
            BeanUtils.copyProperties(processInfo, entity);
            entity.setCreateTime(Java8DateTimeUtils.nowDateTime());
            entity.setJobId(IdHelper.id());
            entity.setRemark("");
            entity.setServer(properties.getUrl());
            return entity;
        });

        jobEntity.setUpdateTime(Java8DateTimeUtils.nowDateTime());
        supervisordRepository.save(jobEntity);
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public List<JobEntity> saveJobList(String instance) throws SupervisordException, IOException {
        Supervisord supervisord = supervisord(instance);
        SupervisordProperties properties = supervisord.getSupervisordProperties();
        List<SupervisordProcess> processList = supervisord.getAllProcessInfo();

        if (!CollectionUtils.isEmpty(processList)) {
            // db中当前实例现有job
            List<JobEntity> dbJobList = supervisordRepository.findByInstanceAndServer(instance, properties.getUrl());

            // 当前实例运行进程job
            List<JobEntity> runJobList = processList.stream()
                    .map(x -> {
                        JobEntity entity = new JobEntity();
                        BeanUtils.copyProperties(x, entity);
                        LocalDateTime now = Java8DateTimeUtils.nowDateTime();
                        entity.setUpdateTime(now);
                        entity.setCreateTime(now);
                        entity.setJobId(IdHelper.id());
                        entity.setInstance(instance);
                        entity.setServer(properties.getUrl());
                        return entity;
                    }).collect(Collectors.toList());

            // 进程运行中的 减去 db现有job 剩下就是需要再次入库的
            List<JobEntity> subtract = runJobList.stream()
                    .filter(b -> dbJobList.stream().map(x -> Tuple.of(x.getInstance(), x.getServer(), x.getName()))
                            .noneMatch(x -> StringUtils.equals(x._1, b.getInstance())
                                    && StringUtils.equals(x._2, b.getServer())
                                    && StringUtils.equals(x._3, b.getName())))
                    .collect(Collectors.toList());

            LOGGER.info("subtract:{}", subtract);

            return supervisordRepository.saveList(subtract);
        }
        return Lists.newArrayList();
    }

    @Override
    public ListUtils.PageWrapper<JobEntity> partition(String instance, int pageNo, int pageSize, String condition) throws SupervisordException, IOException {
        Supervisord supervisord = supervisord(instance);
        SupervisordProperties properties = supervisord.getSupervisordProperties();

        List<SupervisordProcess> processList = supervisord.getAllProcessInfo();

        if (StringUtils.isNotEmpty(condition)) {
            processList = processList.stream()
                    .filter(x -> StringUtils.containsIgnoreCase(x.getName(), condition)).collect(Collectors.toList());
        }

        ListUtils.PageWrapper<SupervisordProcess> partition = ListUtils.partition(pageNo, pageSize, processList);

        List<SupervisordProcess> records = partition.getRecords();

        Set<String> jobNameSet = records.stream().map(SupervisordProcess::getName).collect(Collectors.toSet());
        List<JobEntity> list = supervisordRepository.findByInstanceAndServerAndName(instance, properties.getUrl(), jobNameSet);

        Seq<JobEntity> confSeq = Seq.cast(list.stream(), JobEntity.class);

        Seq<SupervisordProcess> processSeq = Seq.cast(records.stream(), SupervisordProcess.class);
        List<JobEntity> collect = processSeq
                .leftOuterJoin(confSeq, (left, right) -> StringUtils.equals(left.getName(), right.getName()))
                .map(x -> {
                    SupervisordProcess process = x.v1();
                    JobEntity entity = new JobEntity();
                    BeanUtils.copyProperties(process, entity);
                    // entity.setUptime(LocalTime.ofSecondOfDay(process.getNow() - process.getStart()));

                    JobEntity ext = x.v2();
                    if (Objects.nonNull(ext)) {
                        BeanUtils.copyProperties(process, ext);
                        BeanUtils.copyProperties(ext, entity);
                    }
                    return entity;
                }).collect(Collectors.toList());

        return new ListUtils.PageWrapper<>(partition.getPageNo(),
                partition.getPageSize(),
                partition.getTotalSize(),
                collect);
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public JobEntity updateJob(String instance, Long jobId, JobInfoRequest request) {
        JobEntity entity = new JobEntity();
        BeanUtils.copyProperties(request, entity);
        entity.setUpdateTime(Java8DateTimeUtils.nowDateTime());
        return supervisordRepository.save(entity);
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public boolean status(String instance, String processName, String action, boolean wait) throws SupervisordException, IOException {
        Supervisord supervisord = supervisord(instance);
        boolean sign;
        if (StringUtils.equals(action, ACTION_START)) {
            sign = true;
        } else if (StringUtils.equals(action, ACTION_STOP)) {
            sign = false;
        } else {
            throw new BusinessException(ResultCode.PARAM_FAILURE.getMessage(), ResultCode.PARAM_FAILURE.getCode());
        }

        boolean result = sign ? supervisord.startProcess(processName, wait) : supervisord.stopProcess(processName, wait);

        if (result) refreshUpdateTime(instance, processName);

        return result;
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public List<SupervisordProcess> allProcessStatus(String action, String instance, boolean wait) throws SupervisordException, IOException {
        Supervisord supervisord = supervisord(instance);
        List<SupervisordProcess> data;
        switch (action) {
            case SupervisordService.ACTION_START:
                data = supervisord.startAllProcesses(wait);
                break;
            case SupervisordService.ACTION_STOP:
                data = supervisord.stopAllProcesses(wait);
                break;
            default:
                throw new BusinessException();
        }

        if (!CollectionUtils.isEmpty(data)) {
            SupervisordProperties properties = supervisord.getSupervisordProperties();
            List<String> processNameList = data.stream().map(SupervisordProcess::getName).collect(Collectors.toList());
            List<JobEntity> dbJobList = supervisordRepository.findByInstanceAndServerAndName(instance, properties.getUrl(), processNameList);
            List<JobEntity> collect = dbJobList.stream().peek(x -> x.setUpdateTime(Java8DateTimeUtils.nowDateTime())).collect(Collectors.toList());
            supervisordRepository.saveAll(collect);
        }
        return data;
    }


    @Override
    public String readProcessStdoutLog(String instance, ReadLogRequest request) throws IOException, SupervisordException {
        Supervisord supervisord = supervisord(instance);
        return supervisord.readProcessStdoutLog(request.getProcessName(), request.getOffset(), request.getLength());
    }

    @Override
    public String readProcessStderrLog(String instance, ReadLogRequest request) throws IOException, SupervisordException {
        Supervisord supervisord = supervisord(instance);
        return supervisord.readProcessStderrLog(request.getProcessName(), request.getOffset(), request.getLength());
    }

    @Override
    public String tailProcessStdoutLog(String instance, ReadLogRequest request) throws IOException, SupervisordException {
        Supervisord supervisord = supervisord(instance);
        return supervisord.tailProcessStdoutLog(request.getProcessName(), request.getOffset(), request.getLength());
    }

    @Override
    public String tailProcessStderrLog(String instance, ReadLogRequest request) throws IOException, SupervisordException {
        Supervisord supervisord = supervisord(instance);
        return supervisord.tailProcessStderrLog(request.getProcessName(), request.getOffset(), request.getLength());
    }

    @Override
    public boolean clearProcessLog(String instance, String processName) throws SupervisordException, IOException {
        Supervisord supervisord = supervisord(instance);
        return supervisord.clearProcessLogs(processName);
    }

    @Override
    public boolean clearLog(String instance) throws SupervisordException, IOException {
        Supervisord supervisord = supervisord(instance);
        return supervisord.clearLog();
    }

    private Supervisord supervisord(String instance) {
        String beanName = String.join("", instance,
                org.springframework.util.StringUtils.capitalize(SupervisordProperties.SUPERVISORD_PREFIX));
        return SpringContextHolder.getBean(beanName, SupervisordRpc.class);
    }
}
